Lambda 関数を利用して、暗号化されていない EBS を暗号化された EBS に自動で置き換えてみた

Lambda 関数を利用して、暗号化されていない EBS を暗号化された EBS に自動で置き換えてみた

Clock Icon2024.09.09

はじめに

テクニカルサポートの 片方 です。
Lambda 関数を利用して暗号化されていない EBS を、暗号化された EBS に自動で置き換えてみました。
マネジメントコンソールなどで実施する場合は以下の手順が想定されます。

  1. Snapshot を取得
  2. 暗号化を有効にしてSnapshot をコピー
  3. コピーした Snapshot から EBS を作成
  4. 作成した新たな EBS をインスタンスにアタッチ

上記の手順を、Lambada 関数で実施します。なお、ルートボリュームについては対象外です。

https://repost.aws/ja/knowledge-center/ebs-change-encryption-key

実装してみた

以下の順番で実装します。

  • 実行ロール作成
  • Lambda 関数作成

対象の元になる暗号化されていない EBS の設定値や EBS タイプはそのままに、EBS を暗号化して置き換えます。

実行ロール

※ 信頼関係

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

アタッチするポリシー例

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "ec2:DescribeInstances",
                "ec2:DescribeSnapshots",
                "ec2:DescribeVolumes",
                "ec2:CreateSnapshot",
                "ec2:CreateVolume",
                "ec2:AttachVolume",
                "ec2:DetachVolume",
                "ec2:DeleteVolume"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:<region>:<account-id>:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:<region>:<account-id>:log-group:/aws/lambda/<function-name>:*"
            ]
        }
    ]
}

※ 適宜修正してください。

Lambda 関数

Python 3.12 で作成しました。
実行ロールでは、既存のロールを使用するを選択し、先ほど作成したロールを指定します。

実装する Lambda 関数例
import boto3
import time
from botocore.exceptions import ClientError

ec2_client = boto3.client('ec2')

def lambda_handler(event, context):
    # 暗号化されていないEBSボリュームを検索
    unencrypted_volumes = find_unencrypted_volumes()

    if not unencrypted_volumes:
        print("No unencrypted volumes found.")
        return

    for volume in unencrypted_volumes:
        volume_id = volume['VolumeId']
        print(f"Unencrypted volume {volume_id} found.")

        # アタッチされているインスタンスIDを取得
        instance_id = get_attached_instance(volume)

        if not instance_id:
            print(f"Volume {volume_id} is not attached to any instance.")
            continue

        # ルートボリュームでないか確認
        if is_root_volume(volume, instance_id):
            print(f"Skipping root volume {volume_id}.")
            continue

        print(f"Volume {volume_id} is attached to instance {instance_id}. Processing encryption...")

        # AvailabilityZoneとボリュームタイプを取得
        availability_zone = volume['AvailabilityZone']
        volume_type = volume['VolumeType']
        iops = volume.get('Iops')  # IOPSを取得 (io1, io2の場合)

        # スナップショット作成
        snapshot_id = create_snapshot(volume_id)

        # 暗号化された新しいボリュームを作成
        encrypted_volume_id = create_encrypted_volume(snapshot_id, availability_zone, volume_type, iops)

        # 既存のデータボリュームをデタッチ
        detach_volume(instance_id, volume_id)

        # 新しい暗号化されたボリュームをアタッチ
        attach_volume(instance_id, encrypted_volume_id, volume['Attachments'][0]['Device'])

        # 古い未暗号化データボリュームを削除
        delete_volume(volume_id)

def find_unencrypted_volumes():
    # 暗号化されていないボリュームを取得
    response = ec2_client.describe_volumes(Filters=[{'Name': 'encrypted', 'Values': ['false']}])
    return response['Volumes']

def get_attached_instance(volume):
    # アタッチされているインスタンスIDを取得
    if 'Attachments' in volume and volume['Attachments']:
        return volume['Attachments'][0]['InstanceId']
    return None

def is_root_volume(volume, instance_id):
    # ルートデバイスかどうかを確認
    instance_info = ec2_client.describe_instances(InstanceIds=[instance_id])
    root_device_name = instance_info['Reservations'][0]['Instances'][0]['RootDeviceName']
    return volume['Attachments'][0]['Device'] == root_device_name

def create_snapshot(volume_id):
    # スナップショット作成
    response = ec2_client.create_snapshot(VolumeId=volume_id, Description=f"Snapshot of {volume_id}")
    snapshot_id = response['SnapshotId']
    print(f"Snapshot {snapshot_id} created for volume {volume_id}")
    wait_for_snapshot(snapshot_id)
    return snapshot_id

def create_encrypted_volume(snapshot_id, availability_zone, volume_type, iops=None):
    # 暗号化されたボリュームを作成
    create_volume_params = {
        'SnapshotId': snapshot_id,
        'AvailabilityZone': availability_zone,
        'VolumeType': volume_type,
        'Encrypted': True
    }

    # IOPSが必要なボリュームタイプにはIOPSを指定
    if volume_type in ['io1', 'io2'] and iops:
        create_volume_params['Iops'] = iops

    response = ec2_client.create_volume(**create_volume_params)
    encrypted_volume_id = response['VolumeId']
    print(f"Encrypted volume {encrypted_volume_id} created from snapshot {snapshot_id} with type {volume_type}")
    wait_for_volume_state(encrypted_volume_id, 'available')
    return encrypted_volume_id

def detach_volume(instance_id, volume_id):
    # ボリュームをインスタンスからデタッチ
    ec2_client.detach_volume(VolumeId=volume_id, InstanceId=instance_id, Force=True)
    print(f"Detaching volume {volume_id} from instance {instance_id}...")
    wait_for_volume_state(volume_id, 'available')

def attach_volume(instance_id, volume_id, device_name):
    # 新しいボリュームをインスタンスにアタッチ
    ec2_client.attach_volume(VolumeId=volume_id, InstanceId=instance_id, Device=device_name)
    print(f"Attaching volume {volume_id} to instance {instance_id} as {device_name}...")
    check_volume_attached(volume_id)

def check_volume_attached(volume_id):
    # ボリュームがアタッチされるのを待つ
    for attempt in range(10):
        volume = ec2_client.describe_volumes(VolumeIds=[volume_id])['Volumes'][0]
        if volume['State'] == 'in-use':
            print(f"Volume {volume_id} is now in-use state.")
            return
        time.sleep(5)
    raise Exception(f"Volume {volume_id} failed to reach 'in-use' state.")

def delete_volume(volume_id):
    # 古いボリュームを削除
    ec2_client.delete_volume(VolumeId=volume_id)
    print(f"Deleted volume {volume_id}")

def wait_for_volume_state(volume_id, state):
    # ボリュームのステータスを待機
    waiter = ec2_client.get_waiter(f'volume_{state}')
    waiter.wait(VolumeIds=[volume_id])
    print(f"Volume {volume_id} is now in {state} state.")

def wait_for_snapshot(snapshot_id):
    # スナップショットの完了を待機
    waiter = ec2_client.get_waiter('snapshot_completed')
    waiter.wait(SnapshotIds=[snapshot_id])
    print(f"Snapshot {snapshot_id} is now completed.")

※ 適宜修正してください。
なお、暗号化に使用されるキーは、デフォルトで使用される aws/ebs この KMS キーを暗号化に使用します。

https://docs.aws.amazon.com/ja_jp/ebs/latest/userguide/work-with-ebs-encr.html

検証してみた

Window OS のインスタンスを起動して、暗号化されていない EBS ボリュームをアタッチします。
検証の為、全ての EBS タイプをアタッチしてます。

暗号化無し

もちろんフォーマット済みの状態です。

フォーマット済み1
フォーマット済み2

Lambda 関数をテストします。

暫くすると...成功しました!

成功です

EBS のマネジメントコンソール画面を確認します。

暗号化済み

スナップショットも取得されています。
また、デバイス名なども一緒ですので、ご安心ください。成功です!

スクリーンショット 2024-09-07 173648

証跡として残しておりませんが、EBS の中身も(ファイル等)問題なくコピーされてます。

まとめ

もし、タイムアウトで全て完了しない場合は複数回、当該 Lambada 関数を実行するといった対応もご検討ください。すでに暗号化されている EBS とルートボリュームは対象外となるため重複処理はされない認識です。
本ブログが誰かの参考になれば幸いです。

参考資料

アノテーション株式会社について

アノテーション株式会社は、クラスメソッド社のグループ企業として「オペレーション・エクセレンス」を担える企業を目指してチャレンジを続けています。「らしく働く、らしく生きる」のスローガンを掲げ、様々な背景をもつ多様なメンバーが自由度の高い働き方を通してお客様へサービスを提供し続けてきました。現在当社では一緒に会社を盛り上げていただけるメンバーを募集中です。少しでもご興味あれば、アノテーション株式会社WEBサイトをご覧ください。

Share this article

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.